--CREATE TABLE `tbl_star_syslog` (
---`id` BIGINT,
---`ts` BIGINT,
--`trace_id` STRING,
--`oa` STRING,
--`path` STRING,
--`action` INT,
--`update_time` TIMESTAMP,
--`create_time` TIMESTAMP,
-- PRIMARY KEY (id) NOT ENFORCED
---)WITH (
--'connector' = 'mysql-cdc',
--'hostname' = 'test-zhaopin09.db.58dns.org',
--'port' = '53149',
---'database-name' = 'db58_zp_bi',
--'table-name' = 'tbl_star_syslog',
--'username' = 'zpbi_wr','password' = '39086120578ddd85');

create table dwd_zp_resume_post (
    id     BIGINT,
    oper   INT,
	step   int,
	whichMethod string,
	modifiedFields string,
	entity string,
	tjfrom string,
	platform INT,
	addTime BIGINT,
	sink_time string,
	e_time BIGINT
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_resume_post',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_resume_post-agg_test',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_resume_post-QvyDC',
'format' = 'json'
);


CREATE TABLE `tbl_star_syslog_kafka` (
`platform` INT,
`cnt` bigint,
 PRIMARY KEY (platform) NOT ENFORCED
)WITH (
'connector' = 'upsert-kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_tz_resumedelivery',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'tbl_star_syslog_kafka',
-- 'scan.startup.mode' = 'earliest-offset',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_tz_resumedelivery-4h0KC',
'key.format' = 'json',
'value.format' = 'json',
'value.fields-include' = 'ALL'
--'format' = 'debezium-json'
);



INSERT INTO tbl_star_syslog_kafka
SELECT platform, count(*)
FROM dwd_zp_resume_post
group by platform;